<?php
/**
 * Created by PhpStorm.
 * User: BlueIdeaLirui
 * Date: 2019/11/8
 * Time: 18:29
 */

namespace console\controllers\integrate;

use business\base\services\EventService;

use business\company\services\queues\SupplierProducerService;
use business\company\services\queues\SupplierProducerService
    as CompanySupplierProducerService;
use business\bigdata\services\queues\SupplierProducerService
    as BigdataSupplierProducerService;

use business\base\caches\DataCache;

use business\company\services\queues\SupplierConsumerService
    as CompanySupplierConsumerService;
use business\bigdata\services\queues\SupplierConsumerService
    as BigdataSupplierConsumerService;

use common\queue\QueueConstants;

class SupplierController extends ReceiveController
{
    /**
     * 供应商生产者调度
     * php yii integrate/supplier/producer full|increment
     * docker exec -i opensearch_php php /var/www/html/opensearch/protected/yii integrate/supplier/producer full
     * @param $action (full：全量更新索引 increment：增量更新索引)
     */
    public function actionProducer($action)
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $producerClass = [
                'companySupplier' => new CompanySupplierProducerService()
                , 'bigdataSupplier' => new BigdataSupplierProducerService()
            ];

            if (!empty($producerClass)) {
                foreach ($producerClass as $name => $reference) {
                    EventService::register($name, $reference, $action);
                }
            }

            EventService::handler();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 平台供应商生产者调度
     * php yii integrate/supplier/company-producer full|increment
     * @param $action
     */
    public function actionCompanyProducer($action, $incrementTime = 2)
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $producerClass = [
                'companySupplier' => new CompanySupplierProducerService()
            ];

            $data = [];

            if (!empty($producerClass)) {
                foreach ($producerClass as $name => $reference) {
                    EventService::register($name, $reference, $action);
                }
            }

            if ($action == 'increment') {
                $data = [$incrementTime];
            }

            EventService::handler($data);
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 平台供应商生产者调度 测试
     * php yii integrate/supplier/company-producer-test
     * @param $action
     */
    public function actionCompanyProducerTest(){
        echo "\nstart full test...\n";
        (new SupplierProducerService())->fulltest();
        echo "\nend full test...\n";

    }

    /**
     * 大数据供应商生产者调度
     * php yii integrate/supplier/bigdata-producer full|increment
     * @param $action
     */
    public function actionBigdataProducer($action)
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $producerClass = [
                'bigdataSupplier' => new BigdataSupplierProducerService()
            ];

            if (!empty($producerClass)) {
                foreach ($producerClass as $name => $reference) {
                    EventService::register($name, $reference, $action);
                }
            }

            EventService::handler();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 供应商生产者调度
     * @param $action
     */
    public function actionBatchProducer($action)
    {
        try {
            (function($action){
                $producerClass = [
                    'companySupplier' => new CompanySupplierProducerService()
                    , 'bigdataSupplier' => new BigdataSupplierProducerService()
                ];

                if (!empty($producerClass)) {
                    foreach ($producerClass as $name => $reference) {
                        EventService::register($name, $reference, $action);
                    }
                }

                $i = 0;
                $workers = 2;
                echo '开始时间' . microtime() . PHP_EOL;
                $pids = [];

                do {
                    $pids[$i] = pcntl_fork();
                    switch ($pids[$i]) {
                        case -1:
                            echo "fork error : {$i} \r\n";
                            exit;
                        case 0:
                            $i == 0
                                ? EventService::trigger('companySupplier')
                                : EventService::trigger('bigdataSupplier');
                            exit;
                        default:
                            break;
                    }
                    $i++;
                } while ($i < $workers);

                foreach ($pids as $i => $pid) {
                    if($pid) {
                        pcntl_waitpid($pid, $status);
                    }
                }
                echo '结束时间' . microtime()  . PHP_EOL;
            })($action);
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 消费端实现
     * php yii integrate/supplier/consume 'company-supplier-index-queue'|'bigdata-supplier-index-queue' true
     * docker exec -i opensearch_php php /var/www/html/opensearch/protected/yii integrate/supplier/consume 'bigdata-supplier-index-queue' true
     * @param $message
     * @param $messageId
     */
    public function actionProcess($message, $messageId)
    {
        try {
            $cacheKey = $message;
            $module   = 'supplier';
            if (!empty($cacheKey)) {
                $data = (new DataCache())->getRecordByHash(
                    $module . '-' . QueueConstants::QUEUE_CHACHE_NAME, $cacheKey);

                $consumerClass = ['companySupplier' => new CompanySupplierConsumerService()
                    /*, 'bigdataSupplier' => new BigdataSupplierConsumerService()*/];
//                $companySupplierConsumerService = new CompanySupplierConsumerService();
//                $companySupplierConsumerService->process($data['values'], $cacheKey, $messageId);
                if (!empty($consumerClass)) {
                    foreach ($consumerClass as $name => $reference) {
                        EventService::register($name, $reference, 'process');
                    }
                }

                EventService::trigger($data['source'] . 'Supplier', [$data['values'], $cacheKey, $messageId]);
            }
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 重发机制
     */
    public function actionResend()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $producerClass = [
                'companySupplier' => new CompanySupplierProducerService()
                //, 'bigdataSupplier' => new BigdataSupplierProducerService()
            ];

            if (!empty($producerClass)) {
                foreach ($producerClass as $name => $reference) {
                    EventService::register($name, $reference,
                        lcfirst(str_replace('action', '', __FUNCTION__)));
                }
            }

            EventService::handler();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 清除已经消费过的队列数据
     */
    public function actionClearConsumerQueue()
    {
        try {
            echo '开始时间' . microtime() . PHP_EOL;
            $producerClass = [
                'companySupplier' => new CompanySupplierProducerService()
                //, 'bigdataSupplier' => new BigdataSupplierProducerService()
            ];

            if (!empty($producerClass)) {
                foreach ($producerClass as $name => $reference) {
                    EventService::register($name, $reference,
                        lcfirst(str_replace('action', '', __FUNCTION__)));
                }
            }

            EventService::handler();
            echo '结束时间' . microtime()  . PHP_EOL;
        } catch (\Exception $ex){
            \Yii::error($ex->getMessage());
            echo '数据调度出错，错误信息：'.$ex->getMessage();
        }
    }

    /**
     * 接收消息并调用业务处理
     * @param array $messageBody
     */
    protected function successed(\iQueue\MessageBody $messageBody)
    {
        // 获取消息内容
        $message = $messageBody->getMessage();
        $messageId = $messageBody->getMessageId();

        $jsonMsg = (is_array($message) || is_object($message)) ?  json_encode($message) : $message;

        // 执行脚本
        $cmd = sprintf("php %s %s '%s' %s", $this->getYii(), $this->queueAction, $jsonMsg, $messageId);
        $this->debug && print "正在执行:{$cmd}\n";
        $output = system($cmd);
        $this->debug && print "{$output}\n";
    }
}
